package com.edu360.select

import com.edu360.beans.Log
import com.edu360.utils.RptUtils
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object SelectAny {
  def main(args: Array[String]): Unit = {
    // 0 校验参数个数
    if (args.length != 2) {
      println(
        """
          |cn.dmp.report.AreaAnalyseRpt
          |参数：
          | logInputPath
          | resultOutputPath
        """.stripMargin)
      sys.exit()
    }

    // 1 接受程序参数
    val Array(logInputPath, resultOutputPath) = args

    // 2 创建sparkconf->sparkContext
    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")
    // RDD 序列化到磁盘 worker与worker之间的数据传输
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    // sparkConf.registerKryoClasses(Array(classOf[Log]))

    val sc = new SparkContext(sparkConf)

    val filtered = sc.textFile(logInputPath)
      .map(_.split(",", -1))
      .filter(_.length >= 85)

    val cache = filtered.map(arr => {
      val log = Log(arr)

      val req = RptUtils.caculateReq(log.requestmode, log.processnode)
      val rtb = RptUtils.caculateRtb(log.iseffective, log.isbilling, log.isbid, log.adorderid, log.iswin, log.winprice, log.adpayment)
      val showClick = RptUtils.caculateShowClick(log.requestmode, log.iseffective)

      ((log.provincename, log.cityname, log.ispname, log.networkmannername, log.devicetype, log.client, log.appname), req ++ rtb ++ showClick)
      // (省，地市，运营商，网络类型，设备类型，操作系统，媒体,渠道...，List(9个指标数据))
    }).cache()

    //求地域分布
    cache.map(x=>((x._1._1,x._1._2),x._2)).reduceByKey((list1, list2) => {
      list1.zip(list2).map(t => t._1 + t._2)
    }).map(t => t._1._1+","+t._1._2+","+t._2.mkString(","))
      .saveAsTextFile(resultOutputPath)

    //求终端设备运营商
    cache.map(x=>(x._1._3,x._2)).reduceByKey((list1, list2) => {
      list1.zip(list2).map(t => t._1 + t._2)
    }).map(t => t._1+","+t._2.mkString(","))
      .saveAsTextFile(resultOutputPath)

    //求终端设备网络类型
    cache.map(x=>(x._1._4,x._2)).reduceByKey((list1, list2) => {
      list1.zip(list2).map(t => t._1 + t._2)
    }).map(t => t._1+","+t._2.mkString(","))
      .saveAsTextFile(resultOutputPath)

    //求终端设备设备类型
    cache.map(x=>(x._1._5,x._2)).reduceByKey((list1, list2) => {
      list1.zip(list2).map(t => t._1 + t._2)
    }).map(t => t._1+","+t._2.mkString(","))
      .saveAsTextFile(resultOutputPath)

    //求终端设备操作系统
    cache.map(x=>(x._1._4,x._2)).reduceByKey((list1, list2) => {
      list1.zip(list2).map(t => t._1 + t._2)
    }).map(t => t._1+","+t._2.mkString(","))
      .saveAsTextFile(resultOutputPath)

    sc.stop()
  }

}
